#include <stdio.h> 
#include <sys/socket.h> 
#include <sys/time.h>
#include <linux/un.h> 
#include <string.h> 
#include <unistd.h> 
#include "role.h" 
#include <stdlib.h>
#include <string>
#include "test_mgr.h" 

#define SOCKET_PATH "/tmp/test_socket"
#define MAX_BACKLOG 512
#define SEND_INTERVAL 1

#define sk_error(fmt, ...) do { \
    printf("ERROR:file<%s> line<%d> function<%s> ", __FILE__, __LINE__, __func__);\
    printf(fmt, ##__VA_ARGS__);\
    printf("\n");\
} while(0)

static long get_ms_time() {
    struct timeval tv;
    memset(&tv, 0x00, sizeof(tv));

    gettimeofday(&tv, nullptr);
    return tv.tv_sec * 1000 + tv.tv_usec / 1000;
}

struct pipe_client {
    role* server;
    uv_pipe_t pipe;
    void* buffer;
    size_t total;
    size_t index;

    explicit pipe_client(role* from): server(from), buffer(nullptr), total(0), index(0) {
        memset(&pipe, 0x00, sizeof(uv_pipe_t));
        pipe.data = this;
    }
};

struct write_request {
    uv_write_t w;
    role* sr;
    char buf[30];

    explicit write_request(role* s, const std::string& str): sr(s) {
        memset(&w, 0x00, sizeof(uv_write_t));
        w.data = this;
        memset(buf, 0x00, sizeof(buf));
        size_t len = str.size() > 29 ? 29 : str.size();
        str.copy(buf, len);
    }
};

struct write_req {
    uv_write_t w;
    uv_buf_t buf;
};

role::role(uv_loop_t* loop): loop_(loop), type_(role_none), is_connected_(false), send_num_(0) {
    memset(&pipe_, 0x00, sizeof(uv_pipe_t));
    pipe_.data = this;
}

void role::on_pipe_close(uv_handle_t* handle) {
    pipe_client* client = static_cast<pipe_client*>(handle->data);
    if ((uv_handle_t*)&client->pipe != handle) {
        sk_error("unknow error\n");
        return;
    }
    
    if (client->buffer) {
        free(client->buffer);
        client->buffer = nullptr;
        client->total = 0;
        client->index = 0;
    }

    delete client;
}

void role::on_server_alloc(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
    pipe_client* client = static_cast<pipe_client*>(handle->data);
    if ((uv_handle_t*)&client->pipe != handle) {
        sk_error("unknow error\n");
        return;
    }

    if (client->total == 0 || client->buffer == nullptr) {
        client->total = 64;
        client->index = 0;
        client->buffer = malloc(64);

        if (client->buffer == nullptr) {
            perror("server malloc error\n");
            buf->base = nullptr;
            buf->len = 0;
            return;
        }
    }

    if (client->total - client->index < 64) {
        client->buffer = realloc(client->buffer, client->total * 2);

        if (client->buffer == nullptr) {
            perror("server realloc error\n");
            buf->base = nullptr;
            buf->len = 0;
            return;
        }
        client->total *= 2;
    }

    suggested_size = client->total - client->index;
    buf->base = static_cast<char*>(client->buffer) + client->index;
    buf->len = client->total - client->index;
}

void role::on_server_read(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
    pipe_client* client = static_cast<pipe_client*>(handle->data);
    if ((uv_stream_t*)&client->pipe != handle) {
        sk_error("unknow error\n");
        uv_close((uv_handle_t*) &client->pipe, on_pipe_close);
        return;
    }

    if (nread == 0 ) return;
    if (nread == UV_EOF) {
        printf("client closed\n");
        uv_close((uv_handle_t*) &client->pipe, on_pipe_close);
        return;
    }

    if (nread < 0) {
        sk_error("read error:%s from pipe\n", uv_err_name(static_cast<int>(nread)));
        uv_close((uv_handle_t*) &client->pipe, on_pipe_close);
        return;
    }

    if (client->total - client->index < (size_t)nread) {
        sk_error("unknow error");
        uv_close((uv_handle_t*) &client->pipe, on_pipe_close);
        return;
    }
    //printf("server recv:%s\n", std::string(buf->base, nread).c_str());

    client->index += nread;
    size_t len = strlen("echo");
    char* recv_buf = static_cast<char*>(client->buffer);
    while(len <= client->index) {
        if (strncmp(recv_buf, "echo", strlen("echo"))) {
            recv_buf += 1;
            len += 1;
            continue;
        }

        write_request* req = new write_request(client->server, "echo");
        uv_buf_t buffer = uv_buf_init(req->buf,static_cast<unsigned int>(strlen(req->buf)));
        int ret = uv_write(&req->w, handle, &buffer, 1, on_write);
        if (ret != 0) {
            sk_error("write error:%s\n", uv_err_name(ret));
            delete req;
            req = nullptr; 
            return;
        }

        recv_buf = static_cast<char*>(client->buffer) + len;
        len += strlen("echo");
    }

    len -= strlen("echo");
    client->index -= len;
    memmove(client->buffer, recv_buf, client->index);
}

void role::on_client_alloc(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
    role* client = static_cast<role *>(handle->data);
    if ((uv_handle_t*)&client->pipe_ != handle) {
        sk_error("unknow error\n");
        return;
    }

    if (client->total_ == 0 || client->buffer_ == nullptr) {
        client->total_ = 64;
        client->index_ = 0;
        client->buffer_ = malloc(64);

        if (client->buffer_ == nullptr) {
            perror("server malloc error\n");
            buf->base = nullptr;
            buf->len = 0;
            return;
        }
    }

    if (client->total_ - client->index_ < 64) {
        client->buffer_ = realloc(client->buffer_, client->total_ * 2);

        if (client->buffer_ == nullptr) {
            perror("server realloc error\n");
            buf->base = nullptr;
            buf->len = 0;
            return;
        }
        client->total_ *= 2;
    }

    suggested_size = client->total_ - client->index_;
    buf->base = static_cast<char*>(client->buffer_) + client->index_;
    buf->len = client->total_ - client->index_;
}

void role::on_client_read(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
    role* client = static_cast<role *>(handle->data);
    if ((uv_stream_t*)&client->pipe_ != handle) {
        sk_error("unknow error\n");
        return;
    }

    if (nread == 0 ) return;
    if (nread == UV_EOF) {
        sk_error("server closed\n");
        return;
    }

    if (nread < 0) {
        sk_error("read error:%s from pipe\n", uv_err_name(static_cast<int>(nread)));
        return;
    }

    if (client->total_ - client->index_ < (size_t)nread) {
        sk_error("unknow error");
        return;
    }

    //printf("client recv:%s\n", std::string(buf->base, nread).c_str());

    client->index_ += nread;
    size_t len = strlen("echo");
    char* recv_buf = static_cast<char*>(client->buffer_);
    while(len <= client->index_) {
        if (strncmp(recv_buf, "echo", strlen("echo"))) {
            recv_buf += 1;
            len += 1;
            continue;
        }

        if (!client->timestamp_.empty()) {
            long time_interval = get_ms_time() - client->timestamp_.front();
            test_mgr::get()->add_msg_recv_cost(time_interval);
            client->timestamp_.pop_front();
            test_mgr::get()->add_msg_recv_count();
        }
        else
          sk_error("timestamp error\n");

        recv_buf = static_cast<char*>(client->buffer_) + len;
        len += strlen("echo");
    }

    len -= strlen("echo");
    client->index_ -= len;
    memmove(client->buffer_, recv_buf, client->index_);

}

void role::on_connection_cb(uv_stream_t* handle, int status) {
    role* se = static_cast<role*>(handle->data);
    if ((uv_stream_t*)&se->pipe_ != handle) {
        sk_error("unknow error\n");
        return;
    }
    
    pipe_client* client = new pipe_client(se);

    int ret = uv_pipe_init(se->loop_, &client->pipe, 0);
    if (ret != 0) {
        sk_error("pipe init error:%s\n", uv_err_name(ret));
        delete client;
        return;
     }

    ret = uv_accept(handle, (uv_stream_t*)&client->pipe);
    if (ret != 0) {
        sk_error("pipe accept error:%s\n", uv_err_name(ret));
        delete client;
        return;
     }

    ret = uv_read_start((uv_stream_t*)&client->pipe, on_server_alloc, on_server_read);
    if (ret != 0) {
        sk_error("pipe read error:%s\n", uv_err_name(ret));
        uv_close((uv_handle_t*)&client->pipe, on_pipe_close);
        return;
    }
    printf("client connected\n");
}

void role::on_write(uv_write_t* w, int status) {
    write_request* req = static_cast<write_request*>(w->data);
    if (status != 0) {
        sk_error("send msg error:%s\n", uv_err_name(status));
        return;
    }

    delete req;
    req = nullptr;
}

void role::send_message(std::string str) {
    if (type_ == role_client && !is_connected_) {
        sk_error("error: hasn't connected, please connect to server\n");
        return;
    }

    write_request* req = new write_request(this, str);
    uv_buf_t buf = uv_buf_init(req->buf,static_cast<unsigned int>(strlen(req->buf)));
    int ret = uv_write(&req->w, (uv_stream_t*)&pipe_, &buf, 1, on_write);
    if (ret != 0) {
        sk_error("write error:%s\n", uv_err_name(ret));
        delete req;
        req = nullptr; 
        return;
    }
}

int role::create_server() {
    if (type_ != role_server) return -1;
    unlink(SOCKET_PATH);

    int ret = uv_pipe_bind(&pipe_, SOCKET_PATH);
    if (ret != 0) {
        sk_error("pipe bind error:%s\n", uv_err_name(ret));
        return -1;
    }

    ret = uv_listen((uv_stream_t*)&pipe_, MAX_BACKLOG, role::on_connection_cb);
    if (ret != 0) {
        sk_error("pipe listen error:%s\n", uv_err_name(ret));
        return -1;
    }

    return 0;
}

int role::create_client() {
    if (type_ != role_client) return -1;

    struct sockaddr_un addr;
    int sockfd = socket(AF_UNIX, SOCK_STREAM, 0);

    if (sockfd < 0) {
        perror("create client socket error");
        return -1;
    }

    memset(&addr, 0x00, sizeof(addr));
    addr.sun_family = AF_UNIX;
    snprintf(addr.sun_path, sizeof(addr.sun_path), "%s", SOCKET_PATH);

    int ret = connect(sockfd, (const sockaddr*)&addr, sizeof(addr));
    if (ret < 0) {
        perror("connect to server error");
        return -1;
    }

    ret = uv_pipe_open(&pipe_, sockfd);
    if (ret != 0) {
        sk_error("client fd pipe open error:%s\n", uv_err_name(ret));
        return -1;
    }

    is_connected_ = true;

    ret = uv_read_start((uv_stream_t*)&pipe_, on_client_alloc, on_client_read);
    if (ret != 0) {
        sk_error("pipe read error:%s\n", uv_err_name(ret));
        return -1;
    }

    memset(&send_timer_, 0x00, sizeof(send_timer_));
    send_timer_.data = this;

    memset(&send_idle_, 0x00, sizeof(send_idle_));
    send_idle_.data = this;

    timestamp_.clear();

    return 0;
}

int role::start_client_send() {
    int ret = uv_timer_init(loop_, &send_timer_);
    if (ret != 0) {
        sk_error("client timer init error:%s\n", uv_err_name(ret));
        return -1;
    }

    ret = uv_timer_start(&send_timer_, &role::on_client_send_cb, 0, SEND_INTERVAL);
    if (ret != 0) {
        sk_error("client timer start error:%s\n", uv_err_name(ret));
        return -1;
    }

    return 0;
}


int role::start_client_send(int flag) {
    int ret = uv_idle_init(loop_, &send_idle_);
    if (ret != 0) {
        sk_error("client timer init error:%s\n", uv_err_name(ret));
        return -1;
    }
    
    ret = uv_idle_start(&send_idle_, &role::on_client_send_cb);
    if (ret != 0) {
        sk_error("client timer start error:%s\n", uv_err_name(ret));
        return -1;
    }

    return 0;
}

static int stop_timer_handle(uv_timer_t* handle) {
    int ret = uv_timer_stop(handle);
    return ret;
}

static int stop_timer_handle(uv_idle_t* handle) {
    int ret = uv_idle_stop(handle);
    return ret;
}

template<typename T>
void role::on_client_send_cb(T* handle) {
    static int total = test_mgr::get()->msg_per_robot();
    role* client = static_cast<role*>(handle->data);
    
    client->send_message("echo");
    client->timestamp_.push_back(get_ms_time());
    test_mgr::get()->add_msg_send_count();

    //printf("send %d\n", client->send_num_);
    if (++client->send_num_ < total) return;

    int ret = stop_timer_handle(handle);
    if (ret != 0)
        sk_error("send timer stop error");
}

int role::init(role_type type) {
    int ret = 0;
    type_ = type;

    ret = uv_pipe_init(loop_, &pipe_, 0);
    if (ret != 0) {
        sk_error("pipe init error:%s\n", uv_err_name(ret));
        return -1;
    }

    if (type_ == role_server) ret = create_server();

    return ret;
}
